package com.yzkj.siot.kafka;

import cn.hutool.json.JSONUtil;
import com.yzkj.siot.dto.DeviceDataDTO;
import com.yzkj.siot.service.DeviceDataHandler;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;

@Slf4j
@Service
public class KafkaService {

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    @Resource
    private DeviceDataHandler deviceDataHandler;

    public void sendMsg(String topic, String msg) {

        //添加回调
        ListenableFutureCallback<SendResult<String, String>> sendCallBack = new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onFailure(Throwable throwable) {
                log.error("sendMessageAsync failure! topic : {}, message: {}, {}", topic, msg, throwable);
            }

            @Override
            public void onSuccess(SendResult<String, String> stringStringSendResult) {
                log.info("sendMessageAsync success! topic: {}, message: {}", topic, msg);
            }
        };
        int partition = Integer.parseInt(String.valueOf(System.currentTimeMillis() % 2));
        kafkaTemplate.send(topic, partition, "key", msg).addCallback(sendCallBack);
    }

    @KafkaListener(topicPartitions = {@TopicPartition(topic = "${kafka-topic.deviceData}", partitions = {"0"})})
    private void receive0(ConsumerRecord<?, ?> consumer, String message) {
        this.receive(consumer, message);
    }

    @KafkaListener(topicPartitions = {@TopicPartition(topic = "${kafka-topic.deviceData}", partitions = {"1"})})
    private void receive1(ConsumerRecord<?, ?> consumer, String message) {
        this.receive(consumer, message);
    }

    private void receive(ConsumerRecord<?, ?> consumer, String message) {
        log.info("kafka data received topic name => {}, key => {}, partition => {}, offset => {}, message => {}", consumer.topic(), consumer.key(), consumer.partition(), consumer.offset(), message);
        DeviceDataDTO deviceDataDTO = JSONUtil.toBean(message, DeviceDataDTO.class);
        if (deviceDataDTO == null) {
            log.warn("【⚠】 empty kafka msg");
            return;
        }
        deviceDataHandler.dataReceived(deviceDataDTO);
    }
}

